package com.grf.kafka.java.interceptor;

import com.grf.kafka.java.KafkaConstant;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.*;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @author grf
 * @apiNote 消费者-拦截器
 * @since 2024/10/27
 */
@Slf4j
public class InterceptorConsumer {
    public static void main(String[] args) {
        // Part1: 设置发送相关属性
        Properties properties = new Properties();
        // 设置kafka地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConstant.BOOTSTRAP_SERVERS);
        // 每个消费者要指定一个group
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
        // 黄色至key序列化化雷
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 设置value的序列值
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 设置拦截器
        properties.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, "com.grf.kafka.java.interceptor.MyConsumerInterceptor");
        Consumer<String, String> consumer = new KafkaConsumer(properties);
        consumer.subscribe(Arrays.asList(KafkaConstant.TOPIC));
        while (true){
            // PART2:拉取消息
            // 100毫秒超时时间
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofNanos(100));

            // PART3: 处理消息
            for(ConsumerRecord<String, String> record : records){
                // 拦截器中统一添加的信息
                String header = new String(record.headers().lastHeader("test").value(), StandardCharsets.UTF_8);
                log.info("offset={};key={};value={}; heads:{}",record.offset(),record.key(),record.value(), header);
            }
            // 提交offset，消息就不会重复推送
            consumer.commitSync();//同步提交，表示必须等到offset提交后再去拉取下一批消息
            // consumer.commitAsync();//异步提交，表示发送完offset的提交请求后，就开始消费下一批数据了。不用等到broker的确认
        }
    }
}
